package com.ang.controller;

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.ang.base.Constants;
import com.ang.domain.Product;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import cn.hutool.core.date.DateUtil;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;

@Slf4j
@RestController
@RequestMapping("/save")
public class SaveController {
    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 单条数据同步插入
     */
    @RequestMapping("/saveSingle")
    public String saveSingle() throws Exception{
        Product p = new Product(IdUtil.simpleUUID(), "华为P30", "p30", "手机", 100000, 5999.0, DateUtil.date());

        String index = Constants.ES_INDEX_PRODUCT;

        IndexRequest indexRequest = new IndexRequest(index);
        indexRequest.source(JSON.toJSONString(p), XContentType.JSON);
        //设置操作类型，false时，_id相同会覆盖，true会报错
        indexRequest.create(true);
        //设置 _id
        indexRequest.id(p.getId());
        //刷新策略
        indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);

        restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);

        return "success";
    }


    /**
     * 单挑数据异步保存
     */
    @RequestMapping("/saveSingleAsync")
    public String saveSingleAsync(){
        Product p = new Product(IdUtil.simpleUUID(), "小米pad", "mipad", "平板", 100000, 2999.0, DateUtil.date());

        String index = Constants.ES_INDEX_PRODUCT;

        IndexRequest indexRequest = new IndexRequest(index);
        indexRequest.source(JSON.toJSONString(p), XContentType.JSON);
        //设置操作类型，false时，_id相同会覆盖，true会报错
        indexRequest.create(true);
        //设置 _id
        indexRequest.id(p.getId());

        restHighLevelClient.indexAsync(indexRequest, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
                    @Override
                    public void onResponse(IndexResponse indexResponse) {
                        ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                        if (shardInfo.getFailed() > 0) {
                            for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                                log.error("将id为：{}的数据存入ES时存在失败的分片，原因为：{}", indexRequest.id(), failure.getCause());
                            }
                        }
                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.error("{}:存储es时异常，数据信息为", indexRequest.id(), e);
                    }
                }
        );

        return "success";
    }


    /**
     * 批量同步插入
     * 可以将数据设置到BulkProcessor对象中，根据策略批量的刷新到Es中
     */
    @RequestMapping("/saveBatch")
    public void saveBatch() throws IOException {
        Product p1 = new Product(IdUtil.simpleUUID(), "小米pad", "mipad", "平板", 100000, 2999.0, DateUtil.date());
        Product p2 = new Product(IdUtil.simpleUUID(), "华为P40", "p40", "手机", 50000, 4999.0, DateUtil.date());
        Product p3 = new Product(IdUtil.simpleUUID(), "华为Mate60", "mate60", "手机", 100000, 6999.0, DateUtil.date());

        List<Product> dataList = new ArrayList<>(Arrays.asList(p1,p2,p3));

        String index = Constants.ES_INDEX_PRODUCT;

        BulkRequest bulkRequest = new BulkRequest();
        for (Product product : dataList) {
            IndexRequest request = new IndexRequest(index);
            request.source(JSON.toJSONString(product), XContentType.JSON);
            request.opType(DocWriteRequest.OpType.CREATE);//request.create(true);

            bulkRequest.add(request);
        }

        //
        restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
    }
}
